/*
 *
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 *
 */

package org.apache.airavata.metcat.workflowmonitor;

import org.apache.airavata.metcat.dispatcher.DispatchQueue;
import org.apache.airavata.wsmg.client.ConsumerNotificationHandler;
import org.apache.airavata.wsmg.client.MsgBrokerClientException;
import org.apache.airavata.wsmg.client.WseMsgBrokerClient;
import org.apache.axiom.soap.SOAPEnvelope;
import org.apache.axis2.AxisFault;
import org.w3c.dom.Document;


import java.io.ByteArrayOutputStream;


public class WorkflowMonitor {

    private String subscriptionID = "";
    private String BROKER_URL = "";
    private ConsumerNotificationHandler handler;
    private WseMsgBrokerClient api;
    private int consumerPort = 0;
    private String[] consumerServiceEprs;
    private boolean isServerStared = false;

    public WorkflowMonitor(String brokerURL, int port) {
        if (brokerURL != null) {
            this.BROKER_URL = brokerURL;
        }else
        {
            throw new Error("Invalid broker URL property");
        }
        // Is this correct ?
        if (port > 0 && port <= 65535) {
            this.consumerPort = port;
        }else
        {
            throw new Error("Invalid lister port property");
        }
        init();
    }

    private void init() {
        handler = new ConsumerNotificationHandler() {
            public void handleNotification(SOAPEnvelope soapEnvelope) throws AxisFault {
                try {
                    ByteArrayOutputStream out = new ByteArrayOutputStream();
                    Document document = XMLUtils.toDOM(soapEnvelope.getBody().getFirstElement());
                    DispatchQueue.getInstance().addMsgToQueue(document);



                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        };
    }

    public void startMonitor() {
        api = new WseMsgBrokerClient();
        api.init(BROKER_URL);
        try {
            consumerServiceEprs = api.startConsumerService(consumerPort, handler);
            subscriptionID = api.subscribe(consumerServiceEprs[0], ">", null);
            isServerStared = true;
        } catch (MsgBrokerClientException e) {
            e.printStackTrace();
        }
        System.out.println("Started Monitoring for subscription ID: " + subscriptionID);

    }

    public void stopMonitor() {
        if (isServerStared) {
            System.out.println("Stopping Monitoring for subscription ID: " + subscriptionID);
            api.shutdownConsumerService();
        }
    }

    public void destroyMonitor() {
        if (!subscriptionID.equals("")) {
            System.out.println("Destroying workflow monitor for subscription ID: " + subscriptionID);
            try {
                api.unSubscribe(subscriptionID);
            } catch (MsgBrokerClientException e) {
                e.printStackTrace();
            }
        }
    }

    public void destroyMonitor(String consumerURL) {
        if (!subscriptionID.equals("")) {
            System.out.println("Destroying workflow monitor for subscription ID: " + consumerURL);
            try {
                api.unSubscribe(consumerURL);
            } catch (MsgBrokerClientException e) {
                e.printStackTrace();
            }
        }
    }


}


